package com.example.message_queue.mqserver.datacenter;

import com.example.message_queue.common.MqException;
import com.example.message_queue.mqserver.core.Binding;
import com.example.message_queue.mqserver.core.Exchange;
import com.example.message_queue.mqserver.core.MSGQueue;
import com.example.message_queue.mqserver.core.Message;

import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with Intellij IDEA.
 * Description;
 * User:TQ02
 * Data:2024-03-28
 * Time:17:34
 * Text：管理内存的所有数据
 */
public class MemoryDataCenter {
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();

    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
                                //交换机                 队列名     绑定
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>>  bindingsMap = new ConcurrentHashMap<>();

    private ConcurrentHashMap<String, Message> messageMap =new ConcurrentHashMap<>();

    private ConcurrentHashMap<String, LinkedList<Message>> queueMessage = new ConcurrentHashMap<>();


    private ConcurrentHashMap<String, ConcurrentHashMap<String,Message>> queueMessageMapWait =new ConcurrentHashMap<>();

    public void insertExchange(Exchange ex){
        exchangeMap.put(ex.getName(),ex);
        System.out.println("[MemoryDataCenter]:创建了一个："+ex.getName()+" 交换机");
    }
    public Exchange getExchange(String name){
        return exchangeMap.get(name);
    }
    public void deleteExchange(String exName){
        exchangeMap.remove(exName);
        System.out.println("[MemoryDataCenter]:删除了："+exName+" 交换机");

    }

    public void insertQueue(MSGQueue queue){
        queueMap.put(queue.getName(),queue);
        System.out.println("[MemoryDataCenter]:创建了一个："+queue.getName()+" 队列");

    }
    public MSGQueue getQueue(String name){
        return queueMap.get(name);
    }
    public void deleteQueue(String queueName){
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter]:删除了："+queueName+" 队列");

    }

    public void insertBinding(Binding binding) throws MqException {
//        ConcurrentHashMap<String,Binding> bindMap = bindingMap.get(binding.getExchangeName());
//        if(bindMap == null){
//            bindMap = new ConcurrentHashMap<>();
//            bindingMap.put(binding.getExchangeName(),bindMap);
//        }

        ConcurrentHashMap<String,Binding> bindMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
               k -> new ConcurrentHashMap<>());
        synchronized (bindMap){
            if(bindingsMap.get(binding.getQueueName()) != null){
                throw new MqException("[MemoryDataCenter]绑定已经存在");
            }
            bindMap.put(binding.getQueueName(),binding);
        }
        System.out.println("[MemoryDataCenter]绑定成功--交换机:"+binding.getExchangeName()+"  队列名:"+binding.getQueueName());
    }
    // 获取绑定, 写两个版本:
    // 1. 根据 exchangeName 和 queueName 确定唯一一个 Binding
    // 2. 根据 exchangeName 获取到所有的 Bindings
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }
    //获取所有交换机的绑定
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            // 该交换机没有绑定任何队列. 报错.
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName=" + binding.getExchangeName()
                    + ", queueName=" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功! exchangeName=" + binding.getExchangeName()
                + ", queueName=" + binding.getQueueName());
    }

    public void addMessage(Message message){
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息添加成功");
    }
    public Message selectMessage(String MessageId){
       return  messageMap.get(MessageId);
    }
    public void deleteMessage(String MessageName){
        messageMap.remove(MessageName);
        System.out.println("[MemoryDataCenter]消息被移除，MessageName="+MessageName);
    }

    //消息到队列
    public void sendMessage(MSGQueue queue,Message message){
        //把消息放入对应的数据结构种
//        LinkedList<Message> messages = queueMessage.get(queue.getName());
//        if(messages == null){
//            messages.add(message);
//            queueMessage.put(queue.getName(),messages);
//        }
        //简化
        LinkedList<Message> messages = queueMessage.computeIfAbsent(queue.getName(),k -> new LinkedList<>());
        synchronized (messages){
            messages.add(message);
        }
        addMessage(message);
    }

    //队列读取消息
    public Message pollMessage(String queueName){
        LinkedList<Message> messages = queueMessage.computeIfAbsent(queueName,k->new LinkedList<>());
        if(messages == null){
            System.out.println("[MemoryDataCenter]：对列为空");
            return null;
        }
        synchronized (messages){
            if(messages.size() == 0){
                return null;
            }
            //存在头删
            Message curMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter]:消息取出" + curMessage.getMessageId());
            return curMessage;
        }

    }
    //获取队列的消息个数
    public int getMessageCount(String queueName){
        LinkedList<Message> messages = queueMessage.computeIfAbsent(queueName,k->new LinkedList<>());
        if(messages == null){
            System.out.println("[MemoryDataCenter]："+queueName+"对列为空");
            return 0;
        }
        synchronized (messages){
            return messages.size();
        }
    }
   //添加已读未回的消息
    public void addMessageWaitAck(String queueName,Message message){
        ConcurrentHashMap<String,Message> messageConcurrentHashMap = queueMessageMapWait.computeIfAbsent(queueName,
                k->new ConcurrentHashMap<>());
        messageConcurrentHashMap.put(message.getMessageId(),message);
    }
    //删除已确认的消息
    public void removeMessageWaitAck(String name,String messageId){
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageMapWait.get(name);
        if(messageHashMap == null){
            return;
        }
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter]：消息从待确定队列删除 messageId:"+messageId);
    }
    //获取指定的未确认消息
    public Message getMessageWaitAck(String name,String messageId){
        ConcurrentHashMap<String,Message> messageHashMap = queueMessageMapWait.get(name);
        if(messageHashMap == null){
            return null;
        }
        return messageHashMap.get(messageId);
    }
    //将硬盘之前持久化的各个维度的数据恢复到内存
    public void  recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
       exchangeMap.clear();
       queueMap.clear();
       bindingsMap.clear();
       messageMap.clear();
        //恢复所有交换机数据
        List<Exchange> exchanges = diskDataCenter.selectExchange();
        for(Exchange ex:exchanges){
            exchangeMap.put(ex.getName(),ex);
        }
        //恢复所有队列数据
        List<MSGQueue> queues = diskDataCenter.selectQueue();
        for(MSGQueue msg: queues){
            queueMap.put(msg.getName(), msg);
        }
        //恢复所有消息数据
        List<Binding> bindings = diskDataCenter.selectBinding();
        for(Binding binding: bindings){
            ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k->new ConcurrentHashMap<>());
            bindingMap.put(binding.getExchangeName(),binding);
        }
        for(MSGQueue msg:queues){
            LinkedList<Message> message = diskDataCenter.loadAllMessage(msg.getName());
            queueMessage.put(msg.getName(),message);
            for(Message me:message){
                messageMap.put(me.getMessageId(),me);
            }
        }
        // 注意!! 针对 "未确认的消息" 这部分内存中的数据, 不需要从硬盘恢复. 之前考虑硬盘存储的时候, 也没设定这一块.
        // 一旦在等待 ack 的过程中, 服务器重启了, 此时这些 "未被确认的消息", 就恢复成 "未被取走的消息" .
        // 这个消息在硬盘上存储的时候, 就是当做 "未被取走"
    }
}